package com.ce.socket;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.log4j.Logger;

public class BSocketPool extends BsSocketPool implements ISocketPool {
	private static Logger log = Logger.getLogger(BSocketPool.class);
	private DaemonThread daemonThread;
	private ConcurrentHashMap<ISocket, SocketState> sPool;
	public static final Random random = new Random();
	/**
	 * socket的状态，被使用，出于忙
	 */
	private static final int SOCKET_STATUS_BUSY = 1;
	/**
	 * socket的状态，已经失效
	 */
	private static final int SOCKET_STATUS_DEAD = 2;
	
	private static final int SOCKET_STATUS_ACTIVE = 3;
	//仅用于测试
	public Map<ISocket,Long> totalPool = new HashMap<ISocket, Long>();;
	public BSocketPool(String server) {
		super(server);
	}

	public ISocket getSocket() {
		// TODO Auto-generated method stub
		return null;
	}

	public void initialize() {
		// check to see if already initialized
		if ( initialized
				&& ( sPool != null ) ) {
			log.error( "-> trying to initialize an already initialized pool" );
			return;
		}

		// pools
		sPool = new ConcurrentHashMap<ISocket,SocketState>( initConn );

		log.debug( "-> initializing pool with following settings:" );
		log.debug( "-> initial size: " + initConn );
		log.debug( "-> min spare   : " + minConn );
		log.debug( "-> max spare   : " + maxConn );

		// if servers is not set, or it empty, then
		// throw a runtime exception
		if ( server == null || server.length() <= 0 ) {
			log.error( "-> server is illegal" );
			throw new IllegalStateException( "++++ server is illegal" );
		}

		// initalize our internal hashing structures
		populateBuckets();
		
		// mark pool as initialized
		this.initialized = true;

		// start maint thread
		if ( this.daemonSleep > 0 )
			this.startDaemonThread();

	}
	private void populateBuckets() {
		log.debug( "-> initializing internal hashing structure for consistent hashing" );

		// create initial connections
		log.debug( "-> creating initial connections (" + initConn + ") for host: " + server );

		for ( int j = 0; j < initConn; j++ ) {
			ISocket socket = createSocket();
			if ( socket == null ) {
				log.error( "-> failed to create connection to: " + server + " -- only " + j + " created." );
				break;
			}

			addSocketToPool( socket , null, new SocketState(SOCKET_STATUS_ACTIVE, System.currentTimeMillis()), true);
		}
	}
	private ISocket createSocket() {

		ISocket socket = null;

		try {
			socket = new BSocket( server, this.socketTO, this.socketConnectTO, this.nagle );
			totalPool.put(socket, new Long( System.currentTimeMillis() ));
			if ( !socket.isConnected() ) {
				log.error( "-> failed to get DalSocket obj for: " + server + " -- new socket is not connected" );
				addSocketToPool( socket , null, new SocketState(SOCKET_STATUS_DEAD, System.currentTimeMillis()), true);
				socket = null;
			}
			log.info( "-> created and added socket: " + socket.toString() + " for host " + server );
		}
		catch ( Exception ex ) {
			log.error( "-> failed to create DalSocket obj for: " + server +" "+ ex);
			socket = null;
		}
		
		return socket;
	}
	private void checkAllSocket() {
		if(aliveCheck) {
			boolean flag = false;
			for ( Iterator<ISocket> j = sPool.keySet().iterator(); j.hasNext(); ) {
				ISocket socket = j.next();
				boolean isAlive = socket.isAlive();
				flag = flag || isAlive;
				if(!isAlive) {
					addSocketToPool( socket , null, new SocketState(SOCKET_STATUS_DEAD, System.currentTimeMillis()), true);
				} else {
					addSocketToPool( socket , null, new SocketState(sPool.get(socket).status, System.currentTimeMillis()), true);
				}
				log.debug( "-> "+socket+" is status:"+ isAlive);
			}
			poolIsAvail = flag;
		}
	}
	protected boolean addSocketToPool( ISocket socket, SocketState oldValue, SocketState newValue, boolean needReplace ) {
		boolean result = false;
		if ( sPool != null ) {
			if(needReplace) {
				sPool.put(socket, newValue);
				result = true;
			} else {
				return sPool.replace(socket, oldValue, newValue);
			}
		}
		return result;
	}
	
	public byte[] sendData(byte[] data, int retimes) throws Exception {
		ISocket socket = null;
		try {
			if(retimes == 0) {
				log.info("-> retry send data end, failed! retimes:"+retimes);
				throw new DALException("send data failed! retimes:"+retimes);
			}
			socket = getConnection();
			socket.write(data);
			socket.flush();
			log.info("-> retry send data end, succeed! retimes:"+retimes);
			byte[] brep = socket.read();
			recoverySocket(socket, true);
			return brep;
		} catch (DALException e) {
			log.error("-> send time:"+retimes+" error:" , e);
			throw e;
		} catch (Exception e) {
			log.error("-> send time:"+retimes+" error:" , e);
			if(socket != null) {
				recoverySocket(socket, false);
				checkAllSocket();
			}
			return sendData(data, retimes-1);
		}
	}
	protected void recoverySocket( ISocket socket, boolean addToAvail) {
		if(socket == null)
			return;
		log.debug( "-> calling check-in on socket: " + socket.toString() );
		
		if ( socket.isConnected() && addToAvail ) {
			// add to avail pool
			log.debug( "-> returning socket (" + socket.toString() + " to avail pool");
			addSocketToPool( socket , null, new SocketState(SOCKET_STATUS_ACTIVE, System.currentTimeMillis()), true);
		}
		else {
			addSocketToPool( socket , null, new SocketState(SOCKET_STATUS_DEAD, System.currentTimeMillis()), true);
		}
		
	}

	public ISocket getConnection() {
		ISocket socket = null;
		if ( !this.initialized ) {
			log.error( "-> attempting to get SockIO from uninitialized pool!" );
			return null;
		}
		log.info( "->getConnection-1 size/poolIsAvail:("+sPool.size()+"/"+poolIsAvail+")");		
		if ( sPool != null && !sPool.isEmpty() ) {
			int start = (random.nextInt() % sPool.size());
			
			if (start < 0)
				start*= -1;
			
			int count = 0;
			for ( Iterator<ISocket> i = sPool.keySet().iterator(); i.hasNext(); ) {
				if (count < start) {
					i.next();
					count++;
					continue;
				}
				socket = i.next();
				
				if (isFreeSocket(socket,sPool))
					return socket;
			}
			for (Iterator<ISocket> i = sPool.keySet().iterator(); i.hasNext();)
			{
				if (count > 0) {
					socket = i.next();
					if (isFreeSocket(socket, sPool))
						return socket;
					count--;
				} else
					break;
			}
		}
		//check maxCount
		int count = 0;
		for (Iterator<SocketState> iterator = sPool.values().iterator(); iterator.hasNext();) {
			SocketState socketState = iterator.next();
			if(socketState.status != SOCKET_STATUS_DEAD)
				count++;
		}
		if(count >= maxConn)
			return null;
		
		// create one socket -- let the maint thread take care of creating more
		socket = createSocket();
		if (socket != null)
		{
			addSocketToPool(socket, null,new SocketState(SOCKET_STATUS_BUSY, System.currentTimeMillis()), true);
		}

		return socket;
		
	}
	private boolean isFreeSocket(ISocket socket,ConcurrentMap<ISocket, SocketState> pool)
	{	
		if (socket.isConnected())
		{
			if (pool.get(socket).status == SOCKET_STATUS_ACTIVE) {
				if (!addSocketToPool(socket ,pool.get(socket),new SocketState(SOCKET_STATUS_BUSY, System.currentTimeMillis()), false))
					return false;
				// return socket
				return true;

			}

		} else {
			// add to deadpool for later reaping
			addSocketToPool( socket, null, new SocketState(SOCKET_STATUS_DEAD, System.currentTimeMillis()), true);
		}
		
		return false;
	}
	private void startDaemonThread() {

		if ( daemonThread != null ) {

			if ( daemonThread.isRunning() ) {
				log.error( "=== daemon thread already running ===" );
			}
			else {
				daemonThread.start();
			}
		}
		else {
			daemonThread = new DaemonThread( this );
			daemonThread.setInterval( this.daemonSleep );
			daemonThread.start();
		}
	}
	private void stopDaemonThread() {
		if ( daemonThread != null && daemonThread.isRunning() )
			daemonThread.stopThread();
	}
	public void shutDown() {
		log.debug( "=== SockIOPool shutting down... ===" );

		if ( daemonThread != null && daemonThread.isRunning() ) {
			// stop the main thread
			stopDaemonThread();

			// wait for the thread to finish
			while ( daemonThread.isRunning() ) {
				log.debug( "-> waiting for daemon thread to finish run +++" );
				try { Thread.sleep( 500 ); } catch ( Exception ex ) { }
			}
		}

		log.debug( "-> closing all internal pools." );
		closePool();
		daemonThread       = null;
		initialized       = false;
		log.debug( "=== SockIOPool finished shutting down. ===" );
	}

	private void closePool() {
		 for ( Iterator<ISocket> j = sPool.keySet().iterator(); j.hasNext(); ) {
			 ISocket socket = j.next();

			 try {
				 socket.close();
			 }
			 catch ( IOException ioe ) {
				 log.error( "-> failed to trueClose socket: " + socket.toString() + " for host: " + server );
			 }

			 j.remove();
			 socket = null;
		 }
	}

	private void selfDaemon() {
		log.debug( "=== Starting self maintenance.... ===" );

		Integer needSockets = 0;
		int active = 0;
		int busy = 0;
		int dead = 0;
		for (Iterator<SocketState> iterator = sPool.values().iterator(); iterator.hasNext();) {
			SocketState socketState = iterator.next();
			if(socketState.status == SOCKET_STATUS_ACTIVE)
				active++;
			else if(socketState.status == SOCKET_STATUS_BUSY)
				busy++;
			else
				dead++;
		}
		log.info( "->selfDaemon-1 active/busy/dead/poolIsAvail:("+active+"/"+busy+"/"+dead+"/"+poolIsAvail+")");
		// if pool is too small (n < minSpare)
		if ( active + busy < minConn ) {
			// need to create new sockets
			needSockets = minConn - active - busy;
		}

		log.debug( "-> Need to create " + needSockets + " new sockets for pool for host: " + server );

		Set<ISocket> newSock = new HashSet<ISocket>( needSockets );
		for ( int j = 0; j < needSockets; j++ ) {
			ISocket socket = createSocket();

			if ( socket == null )
				break;

			newSock.add( socket );
		}

		// synchronize to add and remove to/from avail pool
		// as well as clean up the busy pool (no point in releasing
		// lock here as should be quick to pool adjust and no
		// blocking ops here)
		int needToClose = 0;
		for ( ISocket socket : newSock )
			addSocketToPool( socket , null, new SocketState(SOCKET_STATUS_ACTIVE, System.currentTimeMillis()), true);

		if ( active + busy > maxConn ) {
			// need to close down some sockets
			int diff        = active + busy - maxConn;
			needToClose = (diff <= poolMultiplier) ? diff : (diff) / poolMultiplier;

			log.debug( "-> need to remove " + needToClose + " spare sockets for pool for host: " + server );

			for ( Iterator<ISocket> j = sPool.keySet().iterator(); j.hasNext(); ) {
				if ( needToClose <= 0 )
					break;

				// remove stale entries
				ISocket socket = j.next();
				addSocketToPool( socket, null, new SocketState(SOCKET_STATUS_DEAD, System.currentTimeMillis()), true);
				needToClose--;
			}
		} else {
			for ( Iterator<ISocket> j = sPool.keySet().iterator(); j.hasNext(); ) {
				// remove stale entries
				ISocket socket = j.next();
				long expire   = sPool.get( socket ).lastTime;

				// if past idle time
				// then close socket
				// and remove from pool
				if ( (expire + maxIdle) < System.currentTimeMillis() ) {
					log.debug( "-> removing stale entry from pool as it is past its idle timeout and pool is over max spare" );
					// remove from the availPool
					addSocketToPool( socket, null, new SocketState(SOCKET_STATUS_DEAD, System.currentTimeMillis()), true);
				}
			}
		}
		

		// go through busy sockets and destroy sockets
		// as needed to maintain pool settings

		// loop through all connections and check to see if we have any hung connections
		for ( Iterator<ISocket> j = sPool.keySet().iterator(); j.hasNext(); ) {
			// remove stale entries
			ISocket socket = j.next();
			if(sPool.get(socket).status == SOCKET_STATUS_BUSY) {
				long hungTime = sPool.get( socket ).lastTime;
	
				// if past max busy time
				// then close socket
				// and remove from pool
				if ( (hungTime + maxBusyTime) < System.currentTimeMillis() ) {
					log.error( "-> removing potentially hung connection from busy pool ... socket in pool for " + (System.currentTimeMillis() - hungTime) + "ms" );
	
					// remove from the busy pool
					addSocketToPool( socket, null, new SocketState(SOCKET_STATUS_DEAD, System.currentTimeMillis()), true);
					j.remove();
				}
			}
		}
		//check all socket in availPool whether avail
		checkAllSocket();
		
		// finally clean out the deadPool

		for ( Iterator<ISocket> j = sPool.keySet().iterator(); j.hasNext();) {
			ISocket socket = j.next();
			try {
				if(sPool.get(socket).status == SOCKET_STATUS_DEAD) {
					socket.close();
					j.remove();
				}
			}
			catch ( Exception ex ) {
				log.error( "-> failed to close DalSocket obj from deadPool " , ex);
				j.remove();
			}
			socket = null;
		}

		log.debug( "=== ending self maintenance. ===" );
	}
	private static class DaemonThread extends Thread {

		// logger
		private static Logger log = Logger.getLogger( DaemonThread.class );

		private BSocketPool pool;
		private long interval      = 1000 * 3; // every 3 seconds
		private boolean stopThread = false;
		private boolean running;

		protected DaemonThread( BSocketPool pool ) {
			this.pool = pool;
			this.setDaemon( true );
			this.setName( "MaintThread" );
		}

		public void setInterval( long interval ) { this.interval = interval; }
		
		public boolean isRunning() {
			return this.running;
		}

		/** 
		 * sets stop variable
		 * and interupts any wait 
		 */
		public void stopThread() {
			this.stopThread = true;
			this.interrupt();
		}

		/** 
		 * Start the thread.
		 */
		public void run() {
			this.running = true;

			while ( !this.stopThread ) {
				try {
					Thread.sleep( interval );

					// if pool is initialized, then
					// run the maintenance method on itself
					if ( pool.isInitialized() )
						pool.selfDaemon();

				}
				catch ( Exception e ) {
					log.error("-> selfMaint error " + e);
				}
			}

			this.running = false;
		}
	}
	
	private class SocketState {
		private int status;
		private long lastTime;
		public SocketState(int status, long lastTime) {
			this.status = status;
			this.lastTime = lastTime;
		}
		public int getStatus() {
			return status;
		}
		public long getLastTime() {
			return lastTime;
		}
	}

	public void check() {
		// TODO Auto-generated method stub
		
	}

	public void setParam(Object... args) {
		// TODO Auto-generated method stub
	}
}
